package com.atguigu.weibo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import static com.atguigu.weibo.Constant.*;

public class WeiBoUtil {

    //获取配置conf
    private static Configuration configuration = HBaseConfiguration.create();
    static {
        configuration.set("hbase.zookeeper.quorum", "192.168.1.131");
    }

    //创建命名空间
    public static void createNamespace(String nameSpace) throws IOException {

        //创建连接
        Connection connection = ConnectionFactory.createConnection(configuration);
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();

        //创建命名空间描述器
        //命名空间类似于关系型数据库中的schema，可以想象成文件夹
        NamespaceDescriptor weiBo = NamespaceDescriptor
                                                .create(nameSpace)
                                                .addConfiguration("creator", "Jinji")
                                                .addConfiguration("create_time", System.currentTimeMillis() + "")
                                                .build();
        //创建操作
        admin.createNamespace(weiBo);

        //关闭资源
        admin.close();
        connection.close();
    }

    //创建表
    public static void createTable(String tableName, int version, String... cfs) throws IOException {

        //创建连接
        Connection connection = ConnectionFactory.createConnection(configuration);
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();

        //创建表描述器
        HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));

        //创建列族
        for (String cf : cfs) {
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(cf);
            columnDescriptor.setMaxVersions(version);

            tableDescriptor.addFamily(columnDescriptor);
        }

        // 根据对表的配置，创建表
        admin.createTable(tableDescriptor);

        //关闭资源
        admin.close();
        connection.close();
    }

    /**
     * 1.更新微博内容表数据
     * 2、更新收件箱表数据
     *      --获取当前操作人的fans
     *      --去往收件箱依次更新数据
     */
    //发布微博
    public static void publishContent(String uid, String content) throws IOException {

        //创建连接
        Connection connection = ConnectionFactory.createConnection(configuration);

        //获取三张操作的表对象
        Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
        Table relationsTable = connection.getTable(TableName.valueOf(TABLE_RELATIONS));
        Table receiveTable = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));


        long timestamp = System.currentTimeMillis();

        //--------------a、微博内容表中添加1条数据---------------
        //组装RowKey
        String rowKey = uid + "_" + timestamp;
        //生成put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("content"), timestamp, Bytes.toBytes(content));

        //往内容表中添加数据
        contentTable.put(put);


        //---------------b、向微博收件箱表中加入发布的RowKey---------------
        //查询用户关系表，得到当前用户有哪些粉丝
        Get get = new Get(Bytes.toBytes(uid));
        get.addFamily(Bytes.toBytes("fans"));
        Result result = relationsTable.get(get);

        //遍历取出当前发布微博的用户的所有粉丝数据
        List<byte[]> fans = new ArrayList<byte[]>();
        for(Cell cell : result.rawCells()){
            fans.add(CellUtil.cloneQualifier(cell));
        }

        //如果该用户没有粉丝，则直接return
        if(fans.size() <= 0) return;

        //开始操作收件箱表
        List<Put> puts = new ArrayList<Put>();
        for(byte[] fan : fans){
            Put fanPut = new Put(fan);
            fanPut.addColumn(Bytes.toBytes("info"), Bytes.toBytes(uid), timestamp, Bytes.toBytes(rowKey));
            puts.add(fanPut);
        }
        receiveTable.put(puts);

        //关闭资源
        connection.close();
    }


    /**
     * 关注用户逻辑
     * a、在微博用户关系表中，对当前主动操作的用户添加新的关注的好友
     * b、在微博用户关系表中，对被关注的用户添加粉丝（当前操作的用户）
     * c、当前操作用户的微博收件箱添加所关注的用户发布的微博rowKey
     */
    public static void addAttends(String uid, String... attends) throws IOException {

        //参数过滤
        if(attends == null || attends.length <= 0 || uid == null || uid.length() <= 0){
            return;
        }

        //创建连接
        Connection connection = ConnectionFactory.createConnection(configuration);

        //获取三张操作的表对象
        Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));
        Table relationsTable = connection.getTable(TableName.valueOf(TABLE_RELATIONS));
        Table receiveTable = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));

        List<Put> puts = new ArrayList<Put>();

        //a、在微博用户关系表中，添加新关注的好友
        Put attendPut = new Put(Bytes.toBytes(uid));
        for(String attend : attends){
            //为当前用户添加关注的人
            attendPut.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend), Bytes.toBytes(attend));

            //b、为被关注的人，添加粉丝
            Put fansPut = new Put(Bytes.toBytes(attend));
            fansPut.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid), Bytes.toBytes(uid));
            //将所有关注的人一个一个的添加到puts（List）集合中
            puts.add(fansPut);
        }
        puts.add(attendPut);
        relationsTable.put(puts);


        //c.1、微博收件箱添加关注的用户发布的微博内容（content）的rowKey
        Scan scan = new Scan();
        //用于存放取出来的关注的人所发布的微博的rowKey
        List<byte[]> rowKeys = new ArrayList<byte[]>();

        for(String attend : attends){
            //过滤扫描rowKey，即：前置位匹配被关注的人的uid_
            RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(attend + "_"));
            //为扫描对象指定过滤规则
            scan.setFilter(filter);
            //通过扫描对象得到scanner
            ResultScanner result = contentTable.getScanner(scan);
            //迭代器遍历扫描出来的结果集
            Iterator<Result> iterator = result.iterator();
            while(iterator.hasNext()){
                //取出每一个符合扫描结果的那一行数据
                Result r = iterator.next();
                for(Cell cell : r.rawCells()){
                    //将得到的rowKey放置于集合容器中
                    rowKeys.add(CellUtil.cloneRow(cell));
                }
            }
        }

        //c.2、将取出的微博rowKey放置于当前操作用户的收件箱中
        if(rowKeys.size() <= 0) return;

        //得到微博收件箱表的操作对象
        //用于存放多个关注的用户的发布的多条微博rowKey信息
        List<Put> recPuts = new ArrayList<Put>();
        for(byte[] rk : rowKeys){
            Put put = new Put(Bytes.toBytes(uid));
            //uid_timestamp
            String rowKey = Bytes.toString(rk);
            //截取uid
            String attendUID = rowKey.substring(0, rowKey.indexOf("_"));
            long timestamp = Long.parseLong(rowKey.substring(rowKey.indexOf("_") + 1));
            //将微博rowKey添加到指定单元格中（注意：这里的时间戳是使用的微博发布时的时间戳）
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attendUID), timestamp, rk);
            recPuts.add(put);
        }

        receiveTable.put(recPuts);

        //关闭资源
        contentTable.close();
        relationsTable.close();
        receiveTable.close();

        connection.close();
    }


    /**
     * 取消关注（remove)
     * a、在微博用户关系表中，对当前主动操作的用户删除对应取关的好友
     * b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人）
     * c、从收件箱中，删除取关的人的微博的rowKey
     */
    public static void removeAttends(String uid, String... attends) throws IOException {

        //过滤数据
        if(uid == null || uid.length() <= 0 || attends == null || attends.length <= 0) return;

        //创建连接
        Connection connection = ConnectionFactory.createConnection(configuration);

        //a、在微博用户关系表中，删除已关注的好友
        Table relationsTable = connection.getTable(TableName.valueOf(TABLE_RELATIONS));

        //待删除的用户关系表中的所有数据
        List<Delete> deletes = new ArrayList<Delete>();
        //当前取关操作者的uid对应的Delete对象
        Delete attendDelete = new Delete(Bytes.toBytes(uid));
        //遍历取关，同时每次取关都要将被取关的人的粉丝-1
        for(String attend : attends){
            attendDelete.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend));
            //b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人）
            Delete fansDelete = new Delete(Bytes.toBytes(attend));
            fansDelete.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid));
            deletes.add(fansDelete);
        }

        deletes.add(attendDelete);
        relationsTable.delete(deletes);

        //c、删除取关的人的微博rowKey 从收件箱表中
        Table receiveTable = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));

        Delete recDelete = new Delete(Bytes.toBytes(uid));
        for(String attend : attends){
            recDelete.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attend));
        }
        receiveTable.delete(recDelete);

        //关闭资源
        relationsTable.close();
        receiveTable.close();

        connection.close();
    }

    /**
     * 获取微博实际内容
     * a、从微博收件箱中获取所有关注的人的发布的微博的rowKey
     * b、根据得到的rowKey去微博内容表中得到数据
     * c、将得到的数据封装到Message对象中
     */
    public static List<Message> getAttendsContent(String uid) throws IOException {

        //创建连接
        Connection connection = ConnectionFactory.createConnection(configuration);

        Table receiveTable = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
        Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT));

        //a、从收件箱中取得微博rowKey
        Get get = new Get(Bytes.toBytes(uid));
        //设置最大版本号
        get.setMaxVersions(5);

        List<byte[]> rowKeys = new ArrayList<byte[]>();
        Result result = receiveTable.get(get);
        for(Cell cell : result.rawCells()){
            rowKeys.add(CellUtil.cloneValue(cell));
        }

        //b、根据取出的所有rowKey去微博内容表中检索数据
        List<Get> gets = new ArrayList<Get>();
        //根据rowKey取出对应微博的具体内容
        for(byte[] rk : rowKeys){
            Get g = new Get(rk);
            gets.add(g);
        }
        //得到所有的微博内容的result对象
        Result[] results = contentTable.get(gets);

        List<Message> messages = new ArrayList<Message>();
        for(Result res : results){
            for(Cell cell : res.rawCells()){

                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String userId = rowKey.substring(0, rowKey.indexOf("_"));
                String timestamp = rowKey.substring(rowKey.indexOf("_") + 1);
                String content = Bytes.toString(CellUtil.cloneValue(cell));

                Message message = new Message();
                message.setContent(content);
                message.setTimestamp(timestamp);
                message.setUid(userId);

                messages.add(message);
            }
        }

        //关闭资源
        receiveTable.close();
        contentTable.close();
        connection.close();

        return messages;
    }

}































